iT邦幫忙

2026 iThome 鐵人賽

DAY 6
0
自我挑戰組

Data Engineer 下班後偷學 AI系列 第 6

用 LangGraph 幫 Pokémon Agent 畫路

  • 分享至 

  • xImage
  •  

LangChain to LangGraph

昨天用 LangChain 的 create_agent 做了一個 Pokémon 組隊 Agent,丟給它幾個 PokéAPI Tools,讓 Model 自己決定要查什麼、什麼時候查,但實作時其實有遇到一個小問題,那就是寶可夢實在是太有名了,Agent 甚至不需要用我給他的 Tools,他其實也可能可以答出來,所以我必須在 System Prompt 裡面循循善誘 Agent 來用我們的工具,難道我沒有辦法「強迫」Agent 執行某些步驟嗎?所以今天打算學一下用 LangGraph 來控制 Workflow ,並升級一下 Pokémon Agent。

其實 LangChain 的 Agent runtime 背後本來就是 LangGraph。去看 create_agent 的回傳型別,會發現它其實是一個 CompiledStateGraph。換句話說,LangChain 的 create_agent 的邏輯是先幫我們把常見的 Agent Workflow (Graph) 給打包好,大概就是:

Model → Tool → Model → Tool → Finish

而直接使用 LangGraph,就是把這層 Workflow 給打開來自己控制。

Graph API

LangGraph 其實有兩種 API 可以選擇,分別是 Graph APIFunctional AIP,它們底層其實是同一個 LangGraph runtime,只是用不同方式描述 Workflow。

Graph API 偏 declarative,直接定義 StateNodeEdge 把 Workflow 結構顯式化;Functional API 則偏 imperative,比較接近一般 Python 寫法,適合原本就有一堆 procedural code,不想為了導入 LangGraph 把全部 function 重構成 Node / Edge。

今天只重點來看 Graph API

State

State 可以理解成整個 Workflow 共用的資料。
每個 Node 都可以讀取目前的 State,處理完之後再更新其中一部分。

可以用 TypedDictPydantic 來定義 Schema:

class BattleState(TypedDict):
    opponents: list[str]
    candidates: list[str]
    selected_team: list[str]
    
graph_builder = StateGraph(BattleState)

Nodes

Node 就是一個步驟,可以是 Python function、API call、Tool、LLM call,什麼都可以,例如:

def find_candidates(state: BattleState):
    candidates = ...

    return {
        "candidates": candidates
    }
    
graph_builder.add_node("find_candidates", find_candidates)

LangGraph 執行到這個 Node 時,會把目前的 State 傳進去。
Node 做完事情後,只要回傳要更新的欄位,LangGraph 再負責把更新合併回整體 State。
假設進來時:

{
    "opponents": ["cloyster", "magneton"],
    "candidates": [],
    "selected_team": [],
}

Node 只回:

{
    "candidates": ["blastoise", "venusaur", "machamp"]
}

LangGraph merge 完之後,新的 State 會變成:

{
    "opponents": ["cloyster", "magneton"],
    "candidates": ["blastoise", "venusaur", "machamp"],
    "selected_team": []
}

Edges

用來描述 Nodes 之間的執行關係,決定下一步去哪。

set_entry_point

Normal Edges

Normal Edges 是固定的,A 做完之後一定會做 B,例如:

graph_builder.add_edge(
    "load_opponents",
    "analyze_weakness",
)

代表:

load_opponents
      ↓
analyze_weakness

再接:

graph_builder.add_edge(
    "analyze_weakness",
    "find_candidates",
)

整條流程就會變成:

load_opponents
      ↓
analyze_weakness
      ↓
find_candidates

Conditional Edges

Conditional Edge 會呼叫一個 function 來決定下一步去哪。

例如候選 Pokémon 不夠,就回去繼續找:

def check_candidates(state: BattleState):
    if len(state["candidates"]) >= 3:
        return "enough"
    return "not_enough"

然後:

graph_builder.add_conditional_edges(
    "find_candidates",
    check_candidates,
    {
        "enough": "select_team",
        "not_enough": "find_more_candidates",
    },
)

可以看到 Conditional Edge 基本上就是根據某個條件來決定下一步要做什麼,等等,不是應該要自己思考和規劃下一步的才叫做 Agent 嗎,那如何才能讓 LLM 替我們決定下一步要做什麼呢?

1. LLM Node 先寫決策到 State,再用 Conditional Edge 路由

可以再狀態中加入一個 next_step

class BattleState(TypedDict):
    opponents: list[str]
    candidates: list[str]
    selected_team: list[str]
    next_step: str

再做一個 LLM Node:

def decide_next_step(state: BattleState):
    response = model.invoke(
        f"""
目前狀態:

對手:
{state["opponents"]}

候選:
{state["candidates"]}

判斷下一步應該做什麼,只能回答:
- inspect_moves
- find_candidates
- compare_stats
- generate_answer
"""
    )

    return {
        "next_step": response.content.strip()
    }

然後 Conditional Edge:

def route(state: BattleState):
    return state["next_step"]


graph_builder.add_conditional_edges(
    "decide_next_step",
    route,
    {
        "inspect_moves": "inspect_moves",
        "find_candidates": "find_candidates",
        "compare_stats": "compare_stats",
        "generate_answer": "generate_answer",
    },
)

2. 直接讓 LLM Node 回傳 Command

LangGraph 的 Command 可以讓一個 Node 同時更新 State + 決定 goto 哪個 Node。官方就把它定位成 dynamic control flow 的 primitive。

舉例來說

from typing import Literal
from langgraph.types import Command


def decide_next_step(state: BattleState) -> Command[
    Literal[
        "inspect_moves",
        "find_candidates",
        "compare_stats",
        "generate_answer",
    ]
]:

    response = model.invoke(
        f"""
根據目前 Pokémon 對戰分析狀態,
決定下一步最需要做什麼。

只能選:
inspect_moves
find_candidates
compare_stats
generate_answer

State:
{state}
"""
    )

    next_step = response.content.strip()

    return Command(
        update={
            "next_step": next_step,
        },
        goto=next_step,
    )

Graph

前面的 StateNodesEdges 都還只是在定義 Workflow,本質上還在 builder 階段,要等到呼叫 compile 之後,LangGraph 才會把這些定義整理成一個真正可執行的 graph。

from langgraph.graph import END, START, StateGraph

graph_builder = StateGraph(BattleState)

graph_builder.add_node("load_opponents", load_opponents)
graph_builder.add_node("find_candidates", find_candidates)

graph_builder.add_edge(START, "load_opponents")
graph_builder.add_edge("load_opponents", "find_candidates")
graph_builder.add_edge("find_candidates", END)

graph = graph_builder.compile()
result = graph.invoke({
    "opponents": ["cloyster", "exeggutor", "magneton"],
})

升級 Pokémon Agent

預計流程:

START
  ↓
resolve_opponents
LLM 判斷使用者輸入的是哪隻第一世代 Pokémon
  ↓
解析成功?
  ├─ No → handle_invalid_opponents → END
  │
  └─ Yes
       ↓
load_opponents
查對手屬性 + Base Stats
       ↓
analyze_weakness
找出能剋制對手的屬性
       ↓
find_candidates
從第一世代 Pokémon 中找候選
       ↓
score_candidates
查所有候選的:
- 屬性
- Base Stats
- 對每個對手的屬性倍率
       ↓
generate_answer
LLM 綜合:
- Matchup 倍率
- Base Stats
- Team Coverage
挑出 3 隻並解釋原因
       ↓
END

程式太冗長了,這邊重點講一下 score_candidates 就好,另外寫一個 get_type_multiplier 來計算攻擊倍率:

def get_type_multiplier(
    attacker_type: str,
    defender_types: list[str],
) -> float:
    """
    計算某個攻擊屬性打在對手所有屬性上的總倍率。
    每次攻擊只有一個 type,但對手本身可以有多個 type

    例如 Fire 打 Water / Ice:
    0.5 × 2 = 1.0
    """

    type_info = get_type_info(attacker_type)

    if "error" in type_info:
        return 1.0

    multiplier = 1.0

    for defender_type in defender_types:
        if defender_type in type_info["double_damage_to"]:
            multiplier *= 2.0
        elif defender_type in type_info["half_damage_to"]:
            multiplier *= 0.5
        elif defender_type in type_info["no_damage_to"]:
            multiplier *= 0.0

    return multiplier


def score_candidates(
    state: BattleState,
) -> dict:
    """
    查詢所有候選 Pokémon 的種族值,
    並計算它們對每個對手的屬性倍率。

    這個 Node 不負責選隊,
    只負責把客觀資料算好交給最後的 LLM。
    """

    opponent_data = state["opponent_data"]
    candidate_scores = []

    # 候選可能很多,平行查詢基本資料
    with ThreadPoolExecutor(max_workers=12) as executor:
        futures = {
            executor.submit(
                get_pokemon,
                name,
            ): name
            for name in state["candidates"]
        }

        for future in as_completed(futures):
            pokemon = future.result()

            if "error" in pokemon:
                continue

            stats = pokemon["stats"]
            base_stat_total = sum(stats.values())

            matchups = {}

            for opponent in opponent_data:
                # 目前還沒有招式資料,
                # 先假設 Pokémon 可以使用自身屬性的攻擊,
                # 計算每個自身屬性對這個對手的倍率。
                type_multipliers = {
                    attacker_type: get_type_multiplier(
                        attacker_type,
                        opponent["types"],
                    )
                    for attacker_type in pokemon["types"]
                }

                matchups[opponent["name"]] = {
                    "best_multiplier": max(
                        type_multipliers.values()
                    ),
                    "type_multipliers": type_multipliers,
                }

            candidate_scores.append({
                "name": pokemon["name"],
                "types": pokemon["types"],
                "stats": stats,
                "base_stat_total": base_stat_total,
                "matchups": matchups,
            })

    # 只為了讓輸出穩定,不代表排名。
    candidate_scores.sort(
        key=lambda pokemon: pokemon["name"]
    )

    return {
        "candidate_scores": candidate_scores
    }

輸出結果:

## 推薦隊伍:**zapdos、pinsir、nidoking**

這個組合以 **zapdos 主攻 cloyster、pinsir 主攻 exeggutor、nidoking 主攻 magneton**,讓三個對手都有明確的屬性優勢對應,同時保留部分交叉 Coverage。

以下只使用你提供的倍率與種族值;**倍率代表自身屬性作為攻擊屬性時的理論相剋,不代表已確認能使用對應招式。**

### 1. 全隊 Coverage

| 推薦 Pokémon | 對 cloyster | 對 exeggutor | 對 magneton | 主要分工 |
|---|---:|---:|---:|---|
| **zapdos** | **2 倍(electric)** | **2 倍(flying)** | 0.5 倍(electric) | 主攻 cloyster,備援 exeggutor |
| **pinsir** | 1 倍(bug) | **4 倍(bug)** | 0.5 倍(bug) | 主攻 exeggutor |
| **nidoking** | 1 倍(poison/ground) | **2 倍(poison)** | **4 倍(ground)** | 主攻 magneton,備援 exeggutor |

**全隊對三個對手的最佳倍率分別為 2/4/4 倍。**  
三個對手都有大於 1 倍的覆蓋;表中的 1 倍只是等倍,不算屬性優勢。

### 2. 各成員的選擇理由

#### **zapdos:負責 cloyster**
- **特攻 125、速度 100、種族值總和 580**。
- electric 對 cloyster 為 **2 倍**。
- cloyster 的**防禦 180、特防 45**,兩者落差很大。因此,在未確認招式的前提下,zapdos 的高特攻是值得優先考慮的輸出潛力;若有合適的特殊招式,更能針對這項差異。
- 速度種族值 **100 > 70**,比 cloyster 高。
- 對 exeggutor 還有 flying **2 倍**,不是只能服務單一對局的選擇。

#### **pinsir:負責 exeggutor**
- **攻擊 125、速度 85、種族值總和 500**。
- bug 對 exeggutor 為 **4 倍**,是提供資料中對該對手的最高倍率。
- 高攻擊搭配 4 倍相剋,提供明確的物理輸出潛力;速度種族值也高於 exeggutor 的 **55**。
- 相較同樣有 bug 4 倍的 scyther,pinsir 的速度較低(85 對 105),但攻擊較高(125 對 110);兩者速度種族值都高於主要目標,因此這裡偏向選擇攻擊較高的 pinsir。
- 它對另外兩個對手沒有屬性優勢,因此由隊友補足,而不是勉強讓它包辦。

#### **nidoking:負責 magneton**
- **攻擊 102、特攻 85、速度 85、種族值總和 505**。
- ground 對 magneton 為 **4 倍**,補上 zapdos 與 pinsir 都只有 0.5 倍的共同進攻缺口。
- 速度種族值 **85 > 70**,高於 magneton。
- 相較 dugtrio,nidoking 速度較低,但攻擊略高(102 對 100),且 HP、防禦、特防分別為 **81/77/75**,高於 dugtrio 的 **35/50/70**。
- 此外,nidoking 對 exeggutor 有 poison **2 倍**;dugtrio 對它則只有 0.5 倍。因此 nidoking 在保留 magneton 4 倍覆蓋的同時,也有更好的交叉 Coverage。

### 3. 為什麼比單純挑最高種族值更適合?

候選中最高種族值是 **dragonite 的 600**,接著是 **articuno、moltres、zapdos 並列 580**。只看總和,容易忽略:

- **dragonite** 對 cloyster 只有 1 倍,對 magneton 只有 0.5 倍。
- **articuno** 同樣對 cloyster 只有 1 倍、對 magneton 只有 0.5 倍,與 dragonite 的優勢目標高度重疊。
- 即使選擇能完整覆蓋的 **dragonite+moltres+zapdos**,全隊對三個對手的最佳倍率仍是 **2/2/2 倍**。

推薦組合則把較低的部分種族值總和,換成:
- 對 exeggutor 的 **4 倍**專門覆蓋;
- 對 magneton 的 **4 倍**專門覆蓋;
- 三名成員清楚且互補的主要分工。

**結論:zapdos+pinsir+nidoking 更符合「倍率、能力值與整體 Coverage 一起考慮」的目標,而不是單純追求總種族值。** 但沒有招式與實際能力配置資料,仍不能據此保證先手、擊倒或勝率。

個人覺得相比 LangChain,LangGraph 最難的地方其實是 State 管理。每個 Node 做完事情後,如果沒有把後續流程需要的資訊更新進 State,那後面的 Node 並不知道前面的 Node 做了什麼,舉例來說因為忘記把寶可夢的中文名稱放進 State,所以最後在回答的時候是輸出英文。

目前 Pokémon Agent 雖然考慮了屬性剋制和種族值,但並未考慮技能,也沒有考慮被攻擊的情況,還有很大的升級空間。

每日一句

昨天相信 Agent 的自律,今天開始寫 Edge。


上一篇
目標讓 LangChain 成為 Pokémon 大師
下一篇
零資料保留 Zero Data Retention (ZDR)
系列文
Data Engineer 下班後偷學 AI8
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言